#include "util/serialize.hxx"
#include "util/io.hxx"
#include "util/fs.hxx"
#include "util/common.hxx"
#include "util/lockguard.hxx"
#include "db/metafile.hxx"



std::string SegmentKey::serialize() {

    StringSerializer ser;

    ser.appendString(m_store);
    ser.appendString(m_partition);
    ser.appendInt(m_segID);

    return ser.str();
}

void SegmentKey::deserialize(const std::string& data) {

    Deserializer des((char*)data.data(), data.size());

    uint32_t len;
    len = des.readInt();
    m_store = des.readString(len);

    len = des.readInt();
    m_partition = des.readString(len);

    m_segID = des.readInt();
}

std::string SegmentKey::toString() {

    InfoSerializer ser;

    ser.appendString("store", m_store);
    ser.appendString("partition", m_partition);
    ser.appendShort("segmentID", m_segID);

    return ser.str();
}

std::string SegmentInfo::serialize() {

    StringSerializer ser;

    ser.appendInt(m_actualDataSize);
    ser.appendInt(m_isAlive);

    return ser.str();
}

void SegmentInfo::deserialize(const std::string& data) {

    Deserializer des((char*)data.data(), data.size());

    m_actualDataSize = des.readInt();
    m_isAlive = des.readInt();
}

std::string SegmentInfo::toString() {

    InfoSerializer ser;

    ser.appendInt("data size", m_actualDataSize);
    ser.appendInt("isAlive", m_isAlive);

    return ser.str();
}

uint64_t MetaFileHeader::MAGIC = 0x9876565430891242;
uint32_t MetaFileHeader::METAFILE_HEADER_LENGTH = 1024;

MetaFileHeader::MetaFileHeader() 
: m_version(0), m_magic(MetaFileHeader::MAGIC) {
}

std::string MetaFileHeader::serialize() {

    StringSerializer ser;

    ser.appendLong(m_version);
    ser.appendLong(m_magic);

    return ser.str();
}

void MetaFileHeader::deserialize(const std::string& data) {

    Deserializer des((char*)data.data(), data.size());

    m_version = des.readLong();
    m_magic = des.readLong();
}

bool MetaFileHeader::checkHeader() {

    return m_magic == MetaFileHeader::MAGIC;
}

std::string MetaFileHeader::toString() {

    InfoSerializer ser;

    ser.appendLong("version", m_version);
    ser.appendLong("magic", m_magic);

    return ser.str();
}

MetaFile* MetaFile::m_instance = NULL;

MetaFile::MetaFile(const std::string& envPath)
: m_envPath(envPath) {

}

MetaFile::~MetaFile() {

    if(m_raf) { delete m_raf; m_raf = NULL; }
    pthread_mutex_destroy(&m_mutex);
}

DbStatus MetaFile::init() {

    DbStatus ret;
    
    int rc = pthread_mutex_init(&m_mutex, NULL);
    if(rc != 0) return DbStatus("MetaFile::init::pthread_mutex_init", rc);

    std::string metaFile = m_envPath + "/meta";
    bool newfile = false;
    if(!FsUtility::fileExists(metaFile)) {

        newfile = true;
    }


    m_raf = new PosixRandomAccessFile(metaFile);
    ret = m_raf->init(); if(!ret.isok()) return ret;


    if(newfile) {

        ret = saveHeader();
    }
    else {

        ret = loadHeader();
        if(ret.getCode() == DbStatus::DB_PARTIALREAD || ret.getCode() ==
                DbStatus::DB_CORRUPTION) {

            ret = saveHeader();
        }

        ret = loadData();
    }

    return ret;
}

DbStatus MetaFile::flush() {

    saveData();

    return m_raf->flush();
}

uint32_t MetaFile::getDataSize(const std::string& store,
                               const std::string& partition,
                               uint16_t segID) {

    SegmentKey key(store, partition, segID);

    uint32_t size = 0;

    LockGuard lock(m_mutex);

    std::map<SegmentKey, SegmentInfo>::iterator iter = m_seginfoMap.find(key);
    if(iter != m_seginfoMap.end()) size = iter->second.m_actualDataSize;

    return size;
}

void MetaFile::setDataSize(const std::string& store,
                           const std::string& partition,
                           uint16_t segID,
                           uint32_t size) {

    SegmentKey key(store, partition, segID);

    LockGuard lock(m_mutex);

    SegmentInfo info; info.m_actualDataSize = size; info.m_isAlive = 1;
    m_seginfoMap[key] = info;
}

uint32_t MetaFile::changeDataSize(const std::string& store,
                                  const std::string& partition,
                                  uint16_t segID,
                                  uint32_t size) {

    SegmentKey key(store, partition, segID);

    LockGuard lock(m_mutex);

    std::map<SegmentKey, SegmentInfo>::iterator iter = m_seginfoMap.find(key);
    int32_t s = 0; bool isalive = true;
    if(iter != m_seginfoMap.end() ) { 
        s = iter->second.m_actualDataSize + size;
        iter->second.m_actualDataSize = s;

        isalive = iter->second.m_isAlive;
        std::map<SegmentKey, SegmentInfo>::iterator iter =
            m_seginfoMap.find(key);
    }
    else {
        s += size;
        SegmentInfo info; info.m_actualDataSize = s; info.m_isAlive = isalive;
        m_seginfoMap[key] = info;
    }

    return s;
}

void MetaFile::removeSeg(const std::string& store,
               const std::string& partition,
               uint16_t segID) {

    SegmentKey key(store, partition, segID);

    LockGuard lock(m_mutex);

    m_seginfoMap.erase(key);
}

bool MetaFile::isAlive(const std::string& store,
                       const std::string& partition,
                       uint16_t segID) {

    SegmentKey key(store, partition, segID);

    LockGuard lock(m_mutex);

    bool isalive = true;
    std::map<SegmentKey, SegmentInfo>::iterator iter = m_seginfoMap.find(key);

    if(iter != m_seginfoMap.end()) isalive = iter->second.m_isAlive; 

    return isalive;
}

void MetaFile::setAlive(const std::string& store,
                        const std::string& partition,
                        uint16_t segID,
                        bool alive) {

    SegmentKey key(store, partition, segID);

    LockGuard lock(m_mutex);

    m_seginfoMap[key].m_isAlive = true;
}

void MetaFile::setLogSync(uint32_t logID, uint32_t synced) {

    LockGuard lock(m_mutex);

    m_logSynced[logID] = synced;
}

void MetaFile::removeLog(uint32_t logID) {

     LockGuard lock(m_mutex);

     m_logSynced.erase(logID);
}

bool MetaFile::isLogSynced(uint32_t logID) {

    LockGuard lock(m_mutex);

    return (m_logSynced.count(logID) > 0 && m_logSynced[logID] == 1)?true:false;
}

DbStatus MetaFile::loadHeader() {

    DbStatus ret;

    std::string header;
    ret = m_raf->read(0, MetaFileHeader::METAFILE_HEADER_LENGTH, header);
    if(!ret.isok()) return ret;

    m_header.deserialize(header);
    bool isvalid = m_header.checkHeader();
    if(!isvalid) return DbStatus("MetaFile::loadHeader", DbStatus::DB_CORRUPTION);

    return ret;
}

DbStatus MetaFile::saveHeader() {

    DbStatus ret;

    std::string header = m_header.serialize();

    ret = m_raf->write(0, header);
    if(!ret.isok()) return ret;

    ret = m_raf->flush();

    return ret; 
}

DbStatus MetaFile::loadData() {

    DbStatus ret;

    uint64_t offset = MetaFileHeader::METAFILE_HEADER_LENGTH; 
    
    //read seg
    for(int i = 0; i < 2; ++i) {
        std::string length; 
        ret = m_raf->read(offset, sizeof(uint32_t), length); 
        if(!ret.isok()) return ret;

        if(length.size() != sizeof(uint32_t)) { 
            ret = DbStatus("MetaFile::loadData", DbStatus::DB_CORRUPTION); 
            return ret;
        }

        offset += sizeof(uint32_t);

        std::string result; uint32_t len = *(uint32_t*)length.data();
        ret = m_raf->read(offset, len, result);
        if(!ret.isok()) return ret;

        if(result.size() != len) ret = DbStatus("MetaFile::loadData", DbStatus::DB_CORRUPTION);

        offset += len;

        if(i == 0) segDeserialize(result);
        else if(i == 1) logDeserialize(result);
    }


    return ret;
}

DbStatus MetaFile::saveData() {

    DbStatus ret;

    uint64_t offset = MetaFileHeader::METAFILE_HEADER_LENGTH;
   
    //save seg 
    std::string resultseg = segSerialize();
    uint32_t len = resultseg.size();
    ret = m_raf->write(offset, (char*)&len, sizeof(uint32_t));
    if(!ret.isok()) return ret;

    offset += sizeof(uint32_t);
    ret = m_raf->write(offset, resultseg);
    offset += len;

    //save log
    std::string resultlog = logSerialize();
    len = resultlog.size();
    ret = m_raf->write(offset, (char*)&len, sizeof(uint32_t));
    if(!ret.isok()) return ret;

    offset += sizeof(uint32_t);
    ret = m_raf->write(offset, resultlog);
    offset += len;

    return ret;
}


std::string MetaFile::segSerialize() {

    StringSerializer ser;

    LockGuard lock(m_mutex);

    for(std::map<SegmentKey, SegmentInfo>::iterator iter = m_seginfoMap.begin(); iter != m_seginfoMap.end(); ++iter) {
   
       std::string key = const_cast<SegmentKey&>(iter->first).serialize(), value = iter->second.serialize();    
       
       ser.appendString(key);
       ser.appendString(value);
    }

    return ser.str();
}

void MetaFile::segDeserialize(const std::string& data) {

    Deserializer des((char*)data.data(), data.size());

    LockGuard lock(m_mutex);

    while(des.curPos() < data.size()) {

        uint32_t len = des.readInt();
        std::string sk = des.readString(len);

        SegmentKey key;
        key.deserialize(sk);

        len = des.readInt();
        std::string si = des.readString(len);

        SegmentInfo seginfo;
        seginfo.deserialize(si);

        m_seginfoMap[key] = seginfo;
    }
}


std::string MetaFile::logSerialize() {

    StringSerializer ser;

    LockGuard lock(m_mutex);

    for(std::map<uint32_t, uint32_t>::iterator iter =  m_logSynced.begin(); iter
            != m_logSynced.end(); ++iter) {

        ser.appendInt(iter->first);
        ser.appendInt(iter->second);
    }

    return ser.str();
}

void MetaFile::logDeserialize(const std::string& data) {

    Deserializer des((char*)data.data(), data.size());

    LockGuard lock(m_mutex);

    while(des.curPos() < data.size()) {
    
        uint32_t logID = des.readInt();
        uint32_t synced = des.readInt();

        m_logSynced[logID] = synced;
    }
}

MetaFile* MetaFile::getInstance(const std::string& envPath) {

    if(m_instance) return m_instance;

    m_instance = new MetaFile(envPath);

    return m_instance;
}

std::string MetaFile::toString() {

    InfoSerializer ser;

    for(std::map<SegmentKey, SegmentInfo>::iterator iter = m_seginfoMap.begin();
            iter != m_seginfoMap.end(); ++iter) {
        ser.appendString("SEGMENT", const_cast<SegmentKey&>(iter->first).toString());
        ser.appendString("SEGMENT INFO", iter->second.toString());
    }

    for(std::map<uint32_t, uint32_t>::iterator iter =  m_logSynced.begin(); iter
            != m_logSynced.end(); ++iter) {
    
        ser.appendInt("logID", iter->first);
        ser.appendInt("synced", iter->second);
    }

    return ser.str();
}
